"""
数据验证和转换工具模块

提供安全的 Pydantic 模型转换功能，带有完整的错误处理和日志记录。
用于将外部数据源（如 DataFrame）转换为内部 Pydantic 模型时的健壮性保证。

作者: 架构组
创建日期: 2025-10-05
"""

import logging
from typing import List, Optional, TypeVar, Type, Callable, Any, Dict
from decimal import Decimal
import pandas as pd
from pydantic import ValidationError

from backend.redfire_framework.monitoring import increment

logger = logging.getLogger(__name__)

# 泛型类型变量
T = TypeVar('T')


def safe_convert_to_model(
    data: Dict[str, Any],
    model_class: Type[T],
    row_identifier: Optional[str] = None,
    skip_on_error: bool = True
) -> Optional[T]:
    """
    安全地将字典数据转换为 Pydantic 模型
    
    Args:
        data: 原始数据字典
        model_class: 目标 Pydantic 模型类
        row_identifier: 行标识符（用于日志，如 "row 123" 或 "symbol=000001.SZ"）
        skip_on_error: 是否在错误时跳过该行（True）还是抛出异常（False）
    
    Returns:
        Optional[T]: 转换后的模型实例，失败时返回 None（如果 skip_on_error=True）
    
    Raises:
        ValidationError: 如果 skip_on_error=False 且验证失败
    
    Example:
        >>> data = {'symbol': '000001.SZ', 'open': 10.5, 'close': 11.0, ...}
        >>> bar = safe_convert_to_model(data, Bar, row_identifier="row 0")
        >>> if bar:
        >>>     bars.append(bar)
    """
    try:
        return model_class(**data)
    except ValidationError as e:
        # 记录详细的验证错误
        error_details = []
        for error in e.errors():
            field = '.'.join(str(loc) for loc in error['loc'])
            msg = error['msg']
            error_details.append(f"{field}: {msg}")
        
        identifier_str = f" ({row_identifier})" if row_identifier else ""
        logger.error(
            f"❌ Pydantic validation failed for {model_class.__name__}{identifier_str}. "
            f"Errors: {'; '.join(error_details)}. "
            f"Data: {data}"
        )
        
        # 记录监控指标
        increment(f"{model_class.__name__.lower()}_validation_errors_total")
        
        if skip_on_error:
            return None
        else:
            raise
    except Exception as e:
        # 捕获其他可能的异常（如类型转换错误）
        identifier_str = f" ({row_identifier})" if row_identifier else ""
        logger.error(
            f"❌ Unexpected error converting to {model_class.__name__}{identifier_str}: {e}. "
            f"Data: {data}",
            exc_info=True
        )
        
        increment(f"{model_class.__name__.lower()}_conversion_errors_total")
        
        if skip_on_error:
            return None
        else:
            raise


def safe_convert_dataframe_to_models(
    df: pd.DataFrame,
    model_class: Type[T],
    row_to_dict_func: Optional[Callable[[pd.Series], Dict[str, Any]]] = None,
    skip_on_error: bool = True
) -> List[T]:
    """
    安全地将 DataFrame 转换为 Pydantic 模型列表
    
    Args:
        df: 源 DataFrame
        model_class: 目标 Pydantic 模型类
        row_to_dict_func: 自定义行转换函数（可选）
                         如果为 None，则使用 row.to_dict()
        skip_on_error: 是否在错误时跳过该行（True）还是抛出异常（False）
    
    Returns:
        List[T]: 成功转换的模型实例列表
    
    Example:
        >>> df = pd.DataFrame([...])
        >>> bars = safe_convert_dataframe_to_models(df, Bar)
        >>> logger.info(f"Successfully converted {len(bars)}/{len(df)} rows")
    """
    models = []
    failed_count = 0
    
    for idx, row in df.iterrows():
        try:
            # 转换行为字典
            if row_to_dict_func:
                data = row_to_dict_func(row)
            else:
                data = row.to_dict()
            
            # 安全转换为模型
            model = safe_convert_to_model(
                data=data,
                model_class=model_class,
                row_identifier=f"row {idx}",
                skip_on_error=skip_on_error
            )
            
            if model:
                models.append(model)
            else:
                failed_count += 1
                
        except Exception as e:
            # 这里捕获的是 row_to_dict_func 可能抛出的异常
            logger.error(
                f"❌ Error processing row {idx}: {e}. "
                f"Row data: {row.to_dict()}",
                exc_info=True
            )
            failed_count += 1
            
            if not skip_on_error:
                raise
    
    # 记录转换统计
    total_rows = len(df)
    success_count = len(models)
    success_rate = (success_count / total_rows * 100) if total_rows > 0 else 0
    
    if failed_count > 0:
        logger.warning(
            f"⚠️ DataFrame to {model_class.__name__} conversion completed: "
            f"{success_count}/{total_rows} successful ({success_rate:.1f}%), "
            f"{failed_count} failed"
        )
    else:
        logger.info(
            f"✅ DataFrame to {model_class.__name__} conversion completed: "
            f"{success_count}/{total_rows} successful (100%)"
        )
    
    return models


def safe_convert_decimal(value: Any, default: Optional[Decimal] = None) -> Optional[Decimal]:
    """
    安全地将值转换为 Decimal
    
    Args:
        value: 要转换的值
        default: 转换失败时的默认值
    
    Returns:
        Optional[Decimal]: 转换后的 Decimal 或默认值
    
    Example:
        >>> price = safe_convert_decimal(row['close'], Decimal('0'))
    """
    if value is None or pd.isna(value):
        return default
    
    try:
        return Decimal(str(value))
    except (ValueError, TypeError, ArithmeticError) as e:
        logger.warning(f"Failed to convert {value} to Decimal: {e}")
        return default


def validate_bar_data(data: Dict[str, Any]) -> bool:
    """
    验证 Bar 数据的基本完整性（在 Pydantic 验证之前）
    
    Args:
        data: Bar 数据字典
    
    Returns:
        bool: 数据是否有效
    
    Example:
        >>> if validate_bar_data(row_dict):
        >>>     bar = Bar(**row_dict)
    """
    required_fields = ['symbol', 'timestamp', 'open', 'high', 'low', 'close']
    
    # 检查必需字段
    for field in required_fields:
        if field not in data or data[field] is None:
            logger.debug(f"Missing or null required field: {field}")
            return False
        
        # 检查 NaN
        if isinstance(data[field], float) and pd.isna(data[field]):
            logger.debug(f"Field {field} is NaN")
            return False
    
    # 检查价格逻辑（high >= low, high >= open, high >= close, low <= open, low <= close）
    try:
        high = float(data['high'])
        low = float(data['low'])
        open_price = float(data['open'])
        close = float(data['close'])
        
        if high < low:
            logger.debug(f"Invalid price: high ({high}) < low ({low})")
            return False
        
        if high < open_price or high < close:
            logger.debug(f"Invalid price: high ({high}) < open ({open_price}) or close ({close})")
            return False
        
        if low > open_price or low > close:
            logger.debug(f"Invalid price: low ({low}) > open ({open_price}) or close ({close})")
            return False
        
        # 检查负数
        if any(v < 0 for v in [high, low, open_price, close]):
            logger.debug("Negative price detected")
            return False
            
    except (ValueError, TypeError, KeyError) as e:
        logger.debug(f"Price validation error: {e}")
        return False
    
    return True


# ==================== 使用示例 ====================

"""
使用示例 1: 在适配器中转换 DataFrame

```python
from app.utils.data_validation import safe_convert_dataframe_to_models
from app.adapters.core.models import Bar

class TushareAdapter(HistoricalDataAdapter):
    async def query_bars(self, symbol: str, start_time: datetime, end_time: datetime, **kwargs) -> List[Bar]:
        # 从 Tushare 获取 DataFrame
        df = await self._fetch_daily_data(symbol, start_time, end_time)
        
        if df.empty:
            return []
        
        # 定义行转换函数
        def row_to_bar_dict(row: pd.Series) -> Dict[str, Any]:
            return {
                'symbol': row['ts_code'],
                'timestamp': pd.to_datetime(row['trade_date']),
                'open': Decimal(str(row['open'])),
                'high': Decimal(str(row['high'])),
                'low': Decimal(str(row['low'])),
                'close': Decimal(str(row['close'])),
                'volume': Decimal(str(row['vol'])),
                'amount': Decimal(str(row['amount'])) if 'amount' in row else None,
                'frequency': '1d'
            }
        
        # 安全转换（自动跳过错误行）
        bars = safe_convert_dataframe_to_models(
            df=df,
            model_class=Bar,
            row_to_dict_func=row_to_bar_dict,
            skip_on_error=True  # 单行错误不影响整批
        )
        
        return bars
```

使用示例 2: 单行转换

```python
from app.utils.data_validation import safe_convert_to_model, validate_bar_data

# 方式1: 直接转换（带错误处理）
bar = safe_convert_to_model(
    data={'symbol': '000001.SZ', 'timestamp': datetime.now(), ...},
    model_class=Bar,
    row_identifier="symbol=000001.SZ",
    skip_on_error=True
)

if bar:
    bars.append(bar)

# 方式2: 先验证再转换
data = {...}
if validate_bar_data(data):
    try:
        bar = Bar(**data)
        bars.append(bar)
    except ValidationError as e:
        logger.error(f"Validation failed: {e}")
```

使用示例 3: 在 InfluxDB Writer 中（已有错误处理）

```python
# 当前 influxdb_writer_service.py 中的代码已经有了错误处理：
async def write_bars(self, bars: List[Bar], measurement: str = "bars") -> None:
    points = []
    for bar in bars:
        try:
            point = bar.to_influx_point()  # 这里已经有 try-catch
            points.append(point)
        except Exception as e:
            logger.error(f"Failed to convert bar to point: {e}")
            self.metrics.failed_writes += 1
            continue  # 跳过错误的 bar
```
"""
